""""
Case Type   : D库常用语法兼容-常用函数兼容
Case Name   : 所有视图嵌套测试
Create At   : 2025-10-10
Owner       : chen-czywj
Description :
    1.使用兼容D库创建数据库
    2.执行不同类型测试
    3.查看视图返回值正确无误
    4.清理环境
Expect      :
    1.成功
    2.成功
    3.查看视图返回值正确无误
    4.清理环境
History     :
"""

import os
import unittest
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from yat.test import Node
from yat.test import macro

log = Logger()
primary_sh = CommonSH('PrimaryDbUser')

class Function_Commonviews_Case0038(unittest.TestCase):
    def setUp(self):
        log.info(f'--------{os.path.basename(__file__)} start--------')
        self.constant = Constant()
        self.dbname_d = Node('PrimaryDbDmode').db_name
        self.dbname = 'test_views_001'
        self.schema_name = 'sys_view_test_01'
        self.f1_name = 'f1_commonviews_038'
        self.f2_name = 'f2_commonviews_038'
        self.tb1_name = 't1_commonviews_038'
        self.tb2_name = 't2_commonviews_038'
        self.tb3_name = 't3_commonviews_038'
        self.tb4_name = 't4_commonviews_038'
        self.tb5_name = 't5_commonviews_038'
        self.tb6_name = 't6_commonviews_038'
        self.tb7_name = 't7_commonviews_038'
        self.te_name = 'te_commonviews_038'
        self.ix1_name = 'i1_commonviews_038'
        self.ix2_name = 'i2_commonviews_038'
        self.ix3_name = 'i3_commonviews_038'
        self.ix4_name = 'i4_commonviews_038'
        self.ix5_name = 'i5_commonviews_038'
        self.se_name = 's_commonviews_038'
        self.v1_name = 'v1_commonviews_038'
        self.v2_name = 'v2_commonviews_038'
        self.mv_name = 'mv_commonviews_038'
        self.sy_name = 'sy_commonviews_038'
        self.tr_name = 'tr_commonviews_038'
        self.pr_name = 'pr_commonviews_038'

    def test_Function_Commonviews(self):
        text = '-------------step1: 使用兼容D库创建数据库; expect: 成功-------------'
        log.info(text)
        msg = primary_sh.check_connection_status(self.constant.START_STATUS_MSG)
        self.assertTrue(msg)
        create_db = f'''drop database if exists {self.dbname};
            create database if not exists {self.dbname} dbcompatibility = 'D';'''
        res = primary_sh.execut_db_sql(create_db, dbname=self.dbname_d)
        log.info(res)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, res)
        load_extension = primary_sh.execut_db_sql('create extension shark;',
                                     dbname=self.dbname)
        log.info(load_extension)
        self.assertIn(self.constant.create_extension_success, load_extension)

        text = '-------------step2: 执行不同类型测试; expect: 成功-------------'
        log.info(text)
        create_tb = f'''create schema {self.schema_name};
                set search_path to {self.schema_name};
drop table if exists {self.tb1_name};
create table {self.tb1_name} (id int primary key, c2 int not null, 
    c3 char(1), c4 text, c5 numeric(10, 2));
create unique index {self.ix1_name} on {self.tb1_name}(c2);
create index {self.ix2_name} on {self.tb1_name}(c3, c2);
create index {self.ix3_name} on {self.tb1_name}(lower(c4));
create index {self.ix4_name} on {self.tb1_name} using hash(c3);
create index {self.ix5_name} on {self.tb1_name} (c5) where c5 > 100.00;
drop table if exists {self.tb2_name};
create table {self.tb2_name} (f_id int primary key, f_name text not null, f_age int default 18);
drop table if exists {self.tb3_name};
create table {self.tb3_name} (f_id int primary key not null, 
    f_c2 int references {self.tb2_name}(f_id), f_salary real check(f_salary > 0));
create table {self.tb4_name} (c1 int, c2 text, c3 char(1), 
    constraint t_stats_col_pk primary key (c1)) with (orientation = column);
create temp table {self.te_name} as select * from {self.tb1_name};
drop view if exists {self.v1_name};
create view {self.v1_name} as select * from {self.tb1_name};
drop view if exists {self.v2_name};
create view {self.v2_name} as select * from {self.tb1_name} where id > 10 with check option;
drop materialized view if exists {self.mv_name};
create materialized view {self.mv_name} as select * from {self.tb2_name};
create sequence if not exists {self.se_name}
    increment by 1
    start with 1
    nocycle;
select {self.se_name}.nextval;
create table {self.tb5_name} (order_id int primary key 
    default nextval('{self.se_name}'), order_date date);
create synonym {self.sy_name} for {self.tb5_name};
create table {self.tb6_name} (c1 int, c2 int, c3 int);
create table {self.tb7_name} (c1 int, c2 int, c3 int);
create or replace function {self.f1_name}() returns trigger as
\$\$
begin
 insert into {self.tb7_name} values (NEW.c1, NEW.c2, NEW.c3);
 return NEW;
end
\$\$ language plpgsql;
create trigger {self.tr_name}
before insert on {self.tb6_name}
for each row
execute procedure {self.f1_name}();
create or replace function {self.f2_name}(a int, b int) returns int as \$\$
begin
 return a - b;
end;
\$\$ language plpgsql;
create or replace procedure {self.pr_name}(in a int, in b int, out c int) as
begin
 c = a + b;
end;'''
        res = primary_sh.execut_db_sql(create_tb, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, res)
        self.assertIn(self.constant.SET_SUCCESS_MSG, res)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS, res)
        self.assertIn('INSERT 0 0', res)
        self.assertIn(self.constant.CREATE_VIEW_SUCCESS_MSG, res)
        self.assertIn('CREATE MATERIALIZED VIEW', res)
        self.assertIn('CREATE SYNONYM', res)
        self.assertIn(self.constant.CREATE_PROCEDURE_SUCCESS_MSG, res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, res)
        self.assertIn(self.constant.CREATE_SEQUENCE_SUCCESS_MSG, res)
        self.assertNotIn('ERROR', res)

        text = '-------------step3: 查看视图返回值正确无误; expect: 返回值正确-------------'
        log.info(text)
        select_sql = 'select * from sys.tables order by object_id desc limit 30;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.tb1_name, res)
        self.assertIn(self.tb2_name, res)
        self.assertIn(self.tb3_name, res)
        self.assertIn(self.tb4_name, res)
        self.assertIn(self.tb5_name, res)
        self.assertIn(self.tb6_name, res)
        self.assertIn(self.tb7_name, res)
        self.assertNotIn(self.te_name, res)
        select_sql = 'select * from sys.views order by object_id desc limit 30;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.v1_name, res)
        self.assertIn(self.v2_name, res)
        self.assertIn(self.mv_name, res)
        select_sql = 'select * from sys.all_columns order by object_id desc limit 50;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        col_list = ['id', 'c1', 'c2', 'c3', 'c4', 'c5', 'f_id', 'f_name', 'f_age', 
            'f_c2', 'f_salary', 'order_id', 'order_date']
        for col in col_list:
            self.assertIn(col, res)
        select_sql = 'select * from sys.columns order by object_id desc limit 50;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        for col in col_list:
            self.assertIn(col, res)
        select_sql = 'select * from sys.indexes order by object_id desc limit 30;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.ix1_name, res)
        self.assertIn(self.ix2_name, res)
        self.assertIn(self.ix3_name, res)
        self.assertIn(self.ix4_name, res)
        self.assertIn(self.ix5_name, res)
        select_sql = 'select * from sys.procedures order by object_id desc limit 30;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.pr_name, res)
        self.assertIn('(1 row)', res)
        select_sql = 'select * from sys.all_objects order by object_id desc limit 30;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.f1_name, res)
        self.assertIn(self.f2_name, res)
        self.assertIn(self.sy_name, res)
        self.assertIn(self.se_name, res)
        select_sql = 'select * from sys.objects order by object_id desc limit 30;'
        res = primary_sh.execut_db_sql(select_sql, dbname=self.dbname)
        log.info(res)
        self.assertNotIn(self.f1_name, res)
        self.assertNotIn(self.f2_name, res)
        self.assertIn(self.sy_name, res)
        self.assertIn(self.se_name, res)

    def tearDown(self):
        text = '------------------step4: 清理环境; expect: 成功------------------'
        log.info(text)
        drop_db = f'''set search_path to {self.schema_name};
            drop procedure {self.pr_name};
            drop function {self.f2_name};
            drop table {self.tb7_name};
            drop table {self.tb6_name};
            drop function {self.f1_name};
            drop synonym {self.sy_name};
            drop table {self.tb5_name};
            drop sequence {self.se_name};
            drop materialized view {self.mv_name};
            drop view {self.v2_name};
            drop view {self.v1_name};
            drop table {self.te_name};
            drop table {self.tb4_name};
            drop table {self.tb3_name};
            drop table {self.tb2_name};
            drop table {self.tb1_name};
            '''
        res = primary_sh.execut_db_sql(drop_db, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.constant.DROP_PROCEDURE_SUCCESS_MSG, res)
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG, res)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, res)
        self.assertIn('DROP SYNONYM', res)
        self.assertIn('DROP MATERIALIZED VIEW', res)
        self.assertIn(self.constant.DROP_SEQUENCE_SUCCESS_MSG, res)
        self.assertIn(self.constant.DROP_VIEW_SUCCESS_MSG, res)
        drop_db = f'''drop schema if exists {self.schema_name};'''
        res = primary_sh.execut_db_sql(drop_db, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, res)
        drop_db = f'''
            drop database if exists {self.dbname};'''
        res = primary_sh.execut_db_sql(drop_db, dbname=self.dbname_d)
        log.info(res)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, res)
